package za.org.jcicada.comms.impl.tcp;

/**
 * TcpCommsBuffer.java
 *
 * @author Simon Cross
 *
 * Created on 8 January 2009, 14:09
 */

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import za.org.jcicada.comms.ICommsBuffer;
import za.org.jcicada.comms.impl.JavaMessageSerializer;
import za.org.jcicada.comms.messages.CommsMessage;

/**
 * A buffering mechanism for a tcpcomms CommsBidirectionalQueue.
 * 
 * A TcpCommsBuffer is fed data from a stream of bytes (e.g. a tcp connection)
 * and adds completed CommsMessages received to a message queue. In addition, it
 * provides a utility method for converting a CommsMessage to a stream of bytes
 * for sending.
 * 
 * Messages are encoded as a four byte length, followed by the message data and
 * finally a four byte sentinel value.
 * 
 * The toByteBuffer method is used to encode messages to a byte buffer for
 * sending. feedByteBuffer processes a chunk of data. Completed messages can be
 * retrieved using popMessage which returns null if there are no completed
 * messages available.
 */
public class TcpCommsBuffer implements ICommsBuffer {

	private static Log logger = LogFactory.getLog(TcpCommsBuffer.class);

	/** Sentinel value used to mark the ends of messages. */
	protected static final int MSG_SENTINEL = 0xA5A5A5A5; // 10100101 10100101

	/** Size of sentinel value in bytes. */
	protected static final int SENTINEL_SIZE = 4;

	/** Size of length value at start of message in bytes. */
	protected static final int LENGTH_SIZE = 4;

	/** Serializer for converting messages to bytes and back again. */
	protected static final JavaMessageSerializer serializer = new JavaMessageSerializer();

	/** Queue for messages. */
	protected ConcurrentLinkedQueue<CommsMessage> messageQueue;

	/** Unprocessed data (byte buffer). */
	protected ByteBuffer chunk;

	/** Unprocessed data (corresponding byte array). */
	protected byte[] chunkArray;

	/**
	 * Size of pending message (or zero if no message is pending). Does not
	 * include sentinel or message length bytes.
	 */
	protected int pendingMsgSize;

	/**
	 * Create a TcpCommsBuffer object.
	 */
	public TcpCommsBuffer() {
		messageQueue = new ConcurrentLinkedQueue<CommsMessage>();
		chunkArray = new byte[getByteBufferSize(0)];
		chunk = ByteBuffer.wrap(chunkArray);
		pendingMsgSize = 0;
	}

	/**
	 * Calculate the size of the byte buffer needed to hold a message of a given
	 * size
	 */
	public static int getByteBufferSize(int msgSize) {
		return LENGTH_SIZE + msgSize + SENTINEL_SIZE;
	}

	/**
	 * Convert a message to a byte buffer with length and sentinel value.
	 */
	@Override
	public ByteBuffer toByteBuffer(CommsMessage message) throws IOException {
		byte[] bytes = serializer.toBytes(message);
		ByteBuffer sendBuffer = ByteBuffer.allocate(getByteBufferSize(bytes.length));
		sendBuffer.putInt(bytes.length);
		sendBuffer.put(bytes);
		sendBuffer.putInt(MSG_SENTINEL);
		return sendBuffer;
	}

	/**
	 * Process a ByteBuffer, adding just the first complete messages (if any) to
	 * the internal message queue.
	 */
	public void feedOneMessage(ByteBuffer buffer) throws IOException {
		while ((chunk.position() < LENGTH_SIZE) && (buffer.hasRemaining())) {
			// try to read
			chunk.put(buffer.get());
		}

		if (chunk.position() < LENGTH_SIZE) {
			// still not enough data to determine the message length
			return;
		}

		if (pendingMsgSize == 0) {
			// read message size and resizes chunk buffer to make it large
			// enough
			// to contain the message
			int position = chunk.position();
			chunk.rewind();
			pendingMsgSize = chunk.getInt();

			byte[] newChunkArray = new byte[getByteBufferSize(pendingMsgSize)];
			ByteBuffer newChunk = ByteBuffer.wrap(newChunkArray);
			chunk.rewind();
			newChunk.put(chunkArray, 0, position);
			chunkArray = newChunkArray;
			chunk = newChunk;
		}

		int readLength = Math.min(buffer.remaining(), chunk.remaining());
		buffer.get(chunkArray, chunk.position(), readLength);
		chunk.position(chunk.position() + readLength);

		if (chunk.remaining() == 0) {
			// complete message received
			chunk.flip();

			if (chunk.getInt() != pendingMsgSize) {
				logger.error("Message completed but size doesn't match that expected.");
				throw new IOException("Message completed but size doesn't match that expected.");
			}

			byte[] data = new byte[pendingMsgSize];
			chunk.get(data);

			if (chunk.getInt() != MSG_SENTINEL) {
				logger.error("Did not find message sentinel.");
				throw new IOException("Message sentinel not found.");
			}

			CommsMessage commsMsg = serializer.fromBytes(data);
			messageQueue.add(commsMsg);

			chunkArray = new byte[getByteBufferSize(0)];
			chunk = ByteBuffer.wrap(chunkArray);
			pendingMsgSize = 0;
		}
	}

	/**
	 * Process a ByteBuffer, adding any messages to the internal message queue.
	 */
	@Override
	public void feedByteBuffer(ByteBuffer buffer) throws IOException {
		while (buffer.hasRemaining()) {
			feedOneMessage(buffer);
		}
	}

	/**
	 * Pop a message from the internal message queue. Return null if the queue
	 * is empty.
	 */
	@Override
	public CommsMessage popMessage() {
		return messageQueue.poll();
	}

}
